
#include <thread>
#include <gtest.h>
#include <zmq.hpp>

// 一个简单zmq测试例子，运行两个线程，通过进程间的zmq，完成消息的收发
// 考虑到zmq的通信，在不同的通信协议情况下，会存在性能不一样的情况，同样需要测试

// zmq可以自由设置REQ/REQ是bind或connect
// zmq在SUB/PUB模式下，两端限定SUB-connect，PUB-bind，且已经PUBLISH的消息无法再追回
// zmq在PULL/PUSH模式下，两端限定

// tcp对服务端可以使用tcp://*:port，但是对客户端需要指定明确的tcp://address:port
// inproc可以使用poller获取信息，需要使用同一个context

// ipc未能完成测试，找不到相应的代码示例

const int kCounter = 10;

// types

typedef std::thread Thread;
typedef zmq::context_t ZmqContext;
typedef zmq::socket_t ZmqSocket;
typedef zmq::message_t ZmqMessage;

enum ErrorCode : int { OK = 0, ERROR = 1 };

// functions

//-----------------------------------------------------------
// tcp req/rep
//-----------------------------------------------------------

int InitClient() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_REQ);

    std::this_thread::sleep_for(std::chrono::milliseconds(10));

    socket.connect("tcp://localhost:5555");
    std::cout << "client start" << std::endl;

    int counter = 0;
    do {
      ZmqMessage request("request", sizeof("request"));
      std::cout << "client send:" << static_cast<char*>(request.data())
                << std::endl;
      socket.send(request);

      ZmqMessage reply;
      socket.recv(&reply);

      std::cout << "client recv:" << static_cast<char*>(reply.data())
                << std::endl;
    } while (counter++ < kCounter);
  } catch (...) {
    std::cout << "client error" << std::endl;
    return ERROR;
  }

  return OK;
}

int InitServer() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_REP);

    socket.bind("tcp://*:5555");
    std::cout << "server start" << std::endl;

    int counter = 0;
    do {
      ZmqMessage request;
      if (socket.recv(&request)) {
        ZmqMessage reply("reply", sizeof("reply"));
        socket.send(reply);
      }

      std::this_thread::sleep_for(std::chrono::milliseconds(1));
      std::cout << "server counter:" << counter << std::endl;
    } while (counter++ < kCounter);
  } catch (...) {
    std::cout << "server error" << std::endl;
    return ERROR;
  }

  return OK;
}

//-----------------------------------------------------------
// tcp pub/sub
//-----------------------------------------------------------

int InitClient2() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_SUB);

    std::this_thread::sleep_for(std::chrono::milliseconds(5));

    socket.connect("tcp://localhost:5555");
    socket.setsockopt(ZMQ_SUBSCRIBE, "message");
    std::cout << "client start" << std::endl;

    int counter = 0;
    do {
      zmq::pollitem_t items[] = {{socket, 0, ZMQ_POLLIN, 0}};

      if (zmq::poll(items, 1, std::chrono::milliseconds(1))) {
        if (items[0].revents & ZMQ_POLLIN) {
          ZmqMessage reply;
          socket.recv(&reply);

          std::cout << "client got:" << static_cast<char*>(reply.data())
                    << std::endl;
        }
      }
    } while (counter++ < kCounter / 3);

  } catch (...) {
    std::cout << "client error" << std::endl;
    return ERROR;
  }
  return OK;
}

int InitServer2() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_PUB);

    socket.bind("tcp://*:5555");
    std::cout << "server start" << std::endl;

    int counter = 0;
    do {
      ZmqMessage message("message", sizeof("message"));
      socket.send(message);

      std::this_thread::sleep_for(std::chrono::milliseconds(1));
      std::cout << "server counter:" << counter << std::endl;
    } while (counter++ < kCounter);

  } catch (...) {
    std::cout << "server error" << std::endl;
    return ERROR;
  }
  return OK;
}

//-----------------------------------------------------------
// inproc pair
//-----------------------------------------------------------

int InitClient3(ZmqContext* context) {
  try {
    ZmqSocket socket(*context, ZMQ_PAIR);

    socket.connect("inproc://step2");
    std::cout << "client start" << std::endl;

    int counter = 0;
    do {
      zmq::pollitem_t items[] = {{socket, 0, ZMQ_POLLIN, 0}};

      if (zmq::poll(items, 1, std::chrono::milliseconds(-1))) {
        if (items[0].revents & ZMQ_POLLIN) {
          ZmqMessage reply;
          socket.recv(&reply);

          std::cout << "client got:" << static_cast<char*>(reply.data())
                    << std::endl;
        }
      }
    } while (counter++ < kCounter / 3);
  } catch (...) {
    std::cout << "client error" << std::endl;
    return ERROR;
  }
  return OK;
}

int InitServer3(ZmqContext* context) {
  try {
    ZmqSocket socket(*context, ZMQ_PAIR);

    socket.bind("inproc://step2");
    std::cout << "server start" << std::endl;

    int counter = 0;
    do {
      ZmqMessage message("message", sizeof("message"));
      socket.send(message);

      std::this_thread::sleep_for(std::chrono::milliseconds(1));
      std::cout << "server counter:" << counter << std::endl;
    } while (counter++ < kCounter / 3);
  } catch (...) {
    std::cout << "server error" << std::endl;
    return ERROR;
  }
  return OK;
}

void CreateServerThreadForFd() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_PUB);

    socket.bind("tcp://*:5555");
    std::cout << "server start" << std::endl;

    int counter = 0;
    do {
      ZmqMessage message("message", sizeof("message"));
      socket.send(message);

      std::this_thread::sleep_for(std::chrono::milliseconds(1000));
      std::cout << "server counter:" << ++counter << std::endl;
    } while (true);
  } catch (...) {
    std::cout << "server error" << std::endl;
  }
}

// 测试从zmq拿到的fd是否可以用于poll，将会作为uWebsocket的传入值，用于轮询
void CreateClientThreadForFd() {
  try {
    ZmqContext context;
    ZmqSocket socket(context, ZMQ_SUB);

    socket.connect("tcp://localhost:5555");
    socket.setsockopt(ZMQ_SUBSCRIBE, "message");
    std::cout << "client start" << std::endl;

    int fd = (-1);
    size_t fd_size = sizeof(fd);
    socket.getsockopt(
        ZMQ_FD, &fd,
        &fd_size);  // http://api.zeromq.org/3-1:zmq-getsockopt
                    // 拿到的fd只能用于轮询，退出关闭，读写操作都不允许

    fd_set fds;
    timeval tv = {0, 1000};

    do {
      FD_ZERO(&fds);  //每次循环都要清空集合，否则不能检测描述符变化
      FD_SET(fd, &fds);

      switch (select(fd + 1, &fds, &fds, NULL, &tv)) {
        case -1:
          exit(-1);
          break;  //不对各种情况进行处理，简单退出
                  //https://linux.die.net/man/2/select
        case 0:
          break;  //再次轮询
        default:
          //测试sock是否可读，即是否网络上有数据
          if (FD_ISSET(fd, &fds)) {
            ZmqMessage reply;
            socket.recv(&reply);
            std::cout << "client got:" << static_cast<char*>(reply.data())
                      << std::endl;
          }
      }
    } while (true);
  } catch (...) {
    std::cout << "server error" << std::endl;
  }
}

int GetFdFromZmqSocketZmq() {
  Thread server(CreateServerThreadForFd);
  Thread client(CreateClientThreadForFd);
  server.join();
  client.join();

  return OK;
}

// TEST(ZmqTest, HandleNoneZeroInput) {
//   // ZmqContext context;

//   // Thread server3(InitServer3, &context);
//   // Thread client3(InitClient3, &context);
//   // server3.join();
//   // client3.join();

//   // Thread server(InitServer);
//   // Thread client(InitClient);
//   // server.join();
//   // client.join();

//   EXPECT_EQ(OK, GetFdFromZmqSocketZmq());
// }